package com.bwie.kafka;

import com.alibaba.fastjson.JSON;
import com.bwie.mapper.LogMapper;
import com.bwie.pojo.Log;
import com.bwie.utils.R;
import com.bwie.vo.MessageVo;
import org.apache.http.HttpRequest;
import org.checkerframework.checker.units.qual.A;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;

import java.util.Date;

/**
 * @Auther:${涛}
 * @Date:2024/8/8 09:50
 * @Description: KafkaConfig
 * @Version 1.0.0
 */
@Configuration
public class KafkaConfig {


    @Autowired
    StringRedisTemplate stringRedisTemplate;

    @Autowired
    private LogMapper logMapper;


    @KafkaListener(topics = "aaa")
    public void kafka(String message, Acknowledgment acknowledgment) {
        MessageVo messageVo = JSON.parseObject(message, MessageVo.class);
        String msgBody = messageVo.getMsgBody();
        String msgId = messageVo.getMsgId();

        if (!stringRedisTemplate.hasKey(msgId)) {
            acknowledgment.acknowledge();
            return;
        }

        Log log = new Log();
        log.setTime(new Date());
        log.setIp(msgId);
        log.setStatus(0);
        logMapper.insert(log);
    }
}


